package com.example.kafka.listener;

import com.alibaba.fastjson.JSON;
import com.example.kafka.domain.*;
import com.example.kafka.service.ICabinetDoorService;
import com.example.kafka.service.MoneyBeltService;
import com.example.kafka.service.PrinteMachineService;
import com.example.kafka.service.ScreenService;
import com.example.kafka.service.impl.IConveryerBeltServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Date;
import java.util.List;
import java.util.Map;


@Component
@Slf4j
public class MessageListener {


    @Autowired
    private ICabinetDoorService cabinetDoorService;

    @Autowired
    private MoneyBeltService moneyBeltService;

    @Autowired
    private IConveryerBeltServiceImpl converyerBeltService;

    @Autowired
    private PrinteMachineService printeMachineService;

    @Autowired
    private ScreenService screenService;


    @KafkaListener(topics = {"system_log"})
    public void listener(ConsumerRecord<String, String> record) {
        //获取消息
        String message = record.value();
        log.info("接收到来自flume的数据：" + message);
        System.out.println(message);
        //消息偏移量
//        long offset = record.offset();

        //过滤掉脏数据，只保存以{开头的
        if (StringUtils.hasLength(message) && message.startsWith("{")) {
            Map map = (Map) JSON.parse(message);
            List<Map<String, Object>> data = (List<Map<String, Object>>) map.get("data");
            data.forEach(tmp -> {
                String name = tmp.get("name").toString();
                Boolean status;
                Integer current;
                Integer voltage;
                Integer machineId;
                String machineName;
                Integer pressure;
                boolean save;
                switch (name) {
                    case "柜门":
                        // 状态
                        status = (Boolean) tmp.get("status");
                        // 电流
                        current = (Integer) tmp.get("current");
                        // 电压
                        voltage = (Integer) tmp.get("voltage");
                        // 取款机唯一标识
                        machineId = (Integer) tmp.get("machineId");
                        // 机器名
                        machineName = (String) tmp.get("machineName");

                        CabinetDoor cabinetDoor = new CabinetDoor(machineId, machineName, new Date(), status, voltage, current);
                        log.info("柜门信息开始持久化...");
                        save = cabinetDoorService.save(cabinetDoor);
                        if (save) {
                            log.info("柜门信息保存成功：" + cabinetDoor);
                        } else {
                            log.warn("柜门信息保存失败：" + cabinetDoor);
                        }
                        break;
                    case "钞带":
                        // 状态
                        status = (Boolean) tmp.get("status");
                        // 压力
                        pressure = (Integer) tmp.get("pressure");
                        // 电压
                        voltage = (Integer) tmp.get("voltage");
                        // 取款机唯一标识
                        machineId = (Integer) tmp.get("machineId");
                        // 机器名
                        machineName = (String) tmp.get("machineName");

                        MoneyBelt moneyBelt = new MoneyBelt(machineId, new Date(), machineName, status, voltage, pressure);
                        log.info("钞带开始持久化...");
                        save = moneyBeltService.save(moneyBelt);
                        if (save) {
                            log.info("钞带信息保存成功：" + moneyBelt);
                        } else {
                            log.warn("钞带信息保存失败：" + moneyBelt);
                        }
                        break;
                    case "显示屏":
                        // 状态
                        status = (Boolean) tmp.get("status");
                        // 电流
                        current = (Integer) tmp.get("current");
                        // 电压
                        voltage = (Integer) tmp.get("voltage");
                        // 取款机唯一标识
                        machineId = (Integer) tmp.get("machineId");
                        // 机器名
                        machineName = (String) tmp.get("machineName");

                        Screen screen = new Screen(machineId, new Date(), machineName, status, voltage, current);
                        log.info("显示屏信息开始持久化...");
                        save = screenService.save(screen);
                        if (save) {
                            log.info("显示屏信息保存成功：" + screen);
                        } else {
                            log.warn("显示屏信息保存失败：" + screen);
                        }
                        break;
                    case "打印机":
                        // 状态
                        status = (Boolean) tmp.get("status");
                        // 压力
                        pressure = (Integer) tmp.get("pressure");
                        // 电压
                        voltage = (Integer) tmp.get("voltage");
                        // 取款机唯一标识
                        machineId = (Integer) tmp.get("machineId");
                        // 机器名
                        machineName = (String) tmp.get("machineName");

                        PrinteMachine printeMachine = new PrinteMachine(machineId, new Date(), machineName, status, voltage, pressure);
                        log.info("打印机开始持久化...");
                        save = printeMachineService.save(printeMachine);
                        if (save) {
                            log.info("打印机信息保存成功：" + printeMachine);
                        } else {
                            log.warn("打印机信息保存失败：" + printeMachine);
                        }
                        break;
                    case "传送带":
                        // 状态
                        status = (Boolean) tmp.get("status");
                        // 压力
                        pressure = (Integer) tmp.get("speed");
                        // 电压
                        voltage = (Integer) tmp.get("voltage");
                        // 取款机唯一标识
                        machineId = (Integer) tmp.get("machineId");
                        // 机器名
                        machineName = (String) tmp.get("machineName");

                        ConveyerBelt conveyerBelt = new ConveyerBelt(machineId, new Date(), machineName, status, voltage, pressure);
                        log.info("传送带开始持久化...");
                        save = converyerBeltService.save(conveyerBelt);
                        if (save) {
                            log.info("传送带信息保存成功：" + conveyerBelt);
                        } else {
                            log.warn("传送带信息保存失败：" + conveyerBelt);
                        }
                        break;
                }
            });
        }


    }

}

